package com.consumer.config;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ShutdownSignalException;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.CacheMode;
import org.springframework.amqp.rabbit.connection.ChannelListener;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.PropertySource;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * 队列配置
 */
@Component
@PropertySource("classpath:rabbitmq.properties")
public class ConsumerRabbitConfig {
	@Value("${spring.rabbitmq.address}") 
	private String address;
	@Value("${spring.rabbitmq.username}") 
	private String userName;
	@Value("${spring.rabbitmq.password}") 
	private String password;
	@Value("${spring.rabbitmq.virtualHost}") 
	private String virtualHost;
    
	@Bean("connectionFactory")
	public ConnectionFactory connectionFactory() {
		CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
		cachingConnectionFactory.setAddresses(address);
		cachingConnectionFactory.setUsername(userName);
		cachingConnectionFactory.setPassword(password);
        cachingConnectionFactory.setVirtualHost(virtualHost);
        cachingConnectionFactory.setCacheMode(CacheMode.CHANNEL);


		ChannelListener channelListener = new ChannelListener() {
		    @Override
		    public void onCreate(Channel channel, boolean b) {
		    	try {
					channel.basicQos(8);
					System.out.println(">>>>>>>>>>>>>>");
				} catch (IOException e) {
					e.printStackTrace();
				}
		    }
		    @Override
		    public void onShutDown(ShutdownSignalException signal) {
				System.out.println(signal.getReason());
		    }
		};
		List<ChannelListener> listeners = new ArrayList<ChannelListener>();
		listeners.add(channelListener);
		cachingConnectionFactory.setChannelListeners(listeners);
		return cachingConnectionFactory;
	}

	@Bean("rabbitListenerContainerFactory")
	public RabbitListenerContainerFactory<SimpleMessageListenerContainer> rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
		SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
		//允许启动消费者的线程数，当ConnectionFactory中配置的CacheMode为CacheMode.CHANNEL时，可能会启动Concurrent数量的channel
		//当ConnectionFactory中配置的CacheMode为CacheMode.CONNECTION时，可能会启动Concurrent数量的CONNECTION
		//如果设置了2，则本系统启动时就会创建两个connection或者channel（根据CacheMode）
		factory.setConcurrentConsumers(2);
		factory.setMaxConcurrentConsumers(10);//允许同时最大消费数量为10
		factory.setReceiveTimeout(10000L);//10秒
		factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);//设置手动提交
		factory.setConnectionFactory(connectionFactory);
		//消息转换器，负责将生产者生产的消息转换成需要的格式
		factory.setMessageConverter(new JsonMessageConverter());
	    return  factory;
	}
}
